package log.function;

import log.job.StockRiseFallCEPJob;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.math.BigDecimal;

// 股票数据源函数示例
public class StockTickSourceFunction implements SourceFunction<StockRiseFallCEPJob.StockTick> {
    private volatile boolean isRunning = true;
    @Override
    public void run(SourceContext<StockRiseFallCEPJob.StockTick> sourceContext) throws Exception {
        while (isRunning) {
            // 这里需要实现从Kafka、Socket或其他数据源读取股票行情数据
            // 示例：模拟生成测试数据
            StockRiseFallCEPJob.StockTick tick = new StockRiseFallCEPJob.StockTick();
            tick.setTradeDate("2025-08-28");
            tick.setStockCode("600000");
            tick.setCurrentPrice(new BigDecimal("10.50"));
            tick.setPrevWindowClosePrice(new BigDecimal("10.00"));
            tick.setLastPrice(new BigDecimal("10.20"));
            tick.setEventTime(System.currentTimeMillis());

            sourceContext.collect(tick);

            try {
                Thread.sleep(1000); // 每秒产生一条数据
            } catch (InterruptedException e) {
                // Restore interrupted state
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
